package com.dark.count;

import backtype.storm.task.OutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseRichBolt;
import backtype.storm.tuple.Tuple;

import java.util.*;

/**
 * Created by tengxue on 16-6-29.
 */
public class ReportBolt extends BaseRichBolt{

    private Map<String,Long> counts=null;

    public void prepare(Map map, TopologyContext topologyContext, OutputCollector outputCollector) {
        counts=new HashMap<String, Long>();
    }

    public void execute(Tuple tuple) {
        String word=tuple.getStringByField("word");
        Long count=tuple.getLongByField("count");
        counts.put(word,count);
    }

    public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {

    }

    @Override
    public void cleanup() {
        System.out.println("---- FINAL COUNTS ----");
        List<String> keys=new ArrayList<String>();
        keys.addAll(counts.keySet());
        Collections.sort(keys);
        for (String key:keys){
            System.out.println(key+" : "+this.counts.get(key));
        }
        System.out.println("-------------------------------");

    }
}
